package com.galeno.day09;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.File;
import java.io.FileWriter;
import java.io.Writer;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2615:59
 * SideOutput测流(支流,旁路输出)
 * 将整个数据量按照类型打上一道多个标签,按照指定的标签筛选出来,再对数据进行处理
 *
 *  需求L将数据中的偶数奇数和字符串打上标签
 */
public class SideOutputDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2222);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.3", 9999);

        //定义标签
        OutputTag<Integer> oddTag = new OutputTag<Integer>("odd-tag"){};
        OutputTag<Integer> evenTag = new OutputTag<Integer>("even-tag"){};
        OutputTag<String> strTag = new OutputTag<String>("odd-tag"){};

        SingleOutputStreamOperator<String> mainStream = lines.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String in, Context ctx, Collector<String> out) throws Exception {

                try {
                    int num = Integer.parseInt(in);
                    if (num % 2 == 0) {
                        ctx.output(evenTag, num);
                    } else {
                        ctx.output(oddTag, num);
                    }
                } catch (Exception e) {
                    System.out.println("转换异常");
                    ctx.output(strTag, in);
                }
                //未打标签的数据
                out.collect(in);

            }
        });
        DataStream<Integer> oddStream = mainStream.getSideOutput(oddTag);
        oddStream.print("odd:");
        DataStream<Integer> strStream = mainStream.getSideOutput(evenTag);
        strStream.print("str:");
        mainStream.print("main");

        Writer fileWriter = new FileWriter(new File("D:/b.txt"),true);
        fileWriter.write(env.getExecutionPlan());
        fileWriter.flush();

        env.execute();


    }
}
